package com.jltx.storage.downloader.util;

import java.io.IOException;
import java.net.SocketException;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.UndeliverableException;
import io.reactivex.functions.Consumer;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;

/**
 * @author jltxseo
 * Created by junlintianxia on 2018/09/26.
 */
public class RxSchedulerUtil {

    /**
     * 利用RxJava实现的主线程、子线程快速切换的方法。
     *
     * @param task
     * @param <T>
     */

    public static <T> void doTask(final RxTask<T> task) {
        doTask(task, 0);
    }

    /**
     * 利用RxJava实现的主线程、子线程快速切换的方法。
     *
     * @param task
     * @param delay 延时时间 毫秒算
     * @param <T>
     */

    public static <T> void doTask(final RxTask<T> task, long delay) {
        doTask(task, delay, TimeUnit.MILLISECONDS);
    }

    /**
     * 利用RxJava实现的主线程、子线程快速切换的方法。
     *
     * @param task
     * @param delay 延时时间
     * @param unit  时间单位
     * @param <T>
     */
    public static <T> void doTask(final RxTask<T> task, long delay, TimeUnit unit) {
        Observable.just(task)
                .delay(delay, unit)
                .doOnNext(new Consumer<RxTask<T>>() {
                    @Override
                    public void accept(RxTask<T> tRxTask) throws Exception {
                        tRxTask.doOnRxIOThread();
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        task.setDisposable(disposable);
                    }
                })
                .subscribe(new Consumer<RxTask>() {
                    @Override
                    public void accept(RxTask rxTask) throws Exception {
                        rxTask.doOnRxUIThread();
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        throwable.printStackTrace();
                    }
                });
    }


    /**
     * fix bug 针对响应式编程，如果链式已经中断了
     * 出现异常：io.reactivex.exceptions.UndeliverableException
     * 出现原因：调用了多次onError，正常来说，出现一次onError会走正常Observer处理，其他的会走Error handling ,可以通过以下捕捉多次的error：
     */
    public static void setRxJavaErrorHandler() {
        if (RxJavaPlugins.getErrorHandler() != null || RxJavaPlugins.isLockdown()) {
            return;
        }
        RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable e) throws Exception {
                if (e instanceof UndeliverableException) {
                    e = e.getCause();
                }

                if ((e instanceof IOException) || (e instanceof SocketException)) {
                    // fine, irrelevant network problem or API that throws on cancellation
                    return;
                }

                if (e instanceof InterruptedException) {
                    // fine, some blocking code was interrupted by a dispose call
                    return;
                }

                if ((e instanceof NullPointerException) || (e instanceof IllegalArgumentException)) {
                    // that's likely a bug in the application
                    Thread.currentThread().getUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), e);
                    return;
                }

                if (e instanceof IllegalStateException) {
                    // that's a bug in RxJava or in a custom operator
                    Thread.currentThread().getUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), e);
                    return;
                }
            }
        });
    }

    /**
     * 不打断链式过程的
     * 结合compset将让rxjava先从io线程到最后的使用主线程
     *
     * @param <T>
     * @return
     */
    public static <T> ObservableTransformer<T, T> applyIO2MainSchedulersTransformer() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

}
